package cn.iocoder.yudao.module.liveprice.kafka.consumer;

import cn.iocoder.yudao.framework.common.enums.metal.MetalTypeEnum;
import cn.iocoder.yudao.module.infra.api.websocket.WebSocketSenderApi;
import cn.iocoder.yudao.module.liveprice.api.LivePriceApi;
import cn.iocoder.yudao.module.liveprice.api.dto.LivePriceRespDTO;
import cn.iocoder.yudao.module.liveprice.config.KafkaTopicProperties;
import cn.iocoder.yudao.module.liveprice.kafka.RtjDataPriceMessage;
import com.alibaba.fastjson.JSONObject;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.KafkaListenerErrorHandler;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.util.*;

@Component
@Slf4j
public class RtjDataPriceConsumer {

    private static final String LIVE_MESSAGE_PRICE = "live_price";

    @Resource
    private KafkaTopicProperties kafkaTopicProperties;

    // 需要缓存到Redis的代码集合
    private static final Set<String> CACHE_CODES = Set.of("rtjau", "rtjag", "rtjpt", "rtjpd");

    @Resource
    private LivePriceApi livePriceApi;
    @Resource
    private WebSocketSenderApi webSocketSenderApi;

    // 代码到金属类型的映射
    private static final Map<String, Integer> CODE_TO_METAL_TYPE = new HashMap<>();
    static {
        CODE_TO_METAL_TYPE.put("rtjau", MetalTypeEnum.GOLD.getType());
        CODE_TO_METAL_TYPE.put("rtjag", MetalTypeEnum.SILVER.getType());
        CODE_TO_METAL_TYPE.put("rtjpt", MetalTypeEnum.PLATINUM.getType());
        CODE_TO_METAL_TYPE.put("rtjpd", MetalTypeEnum.PALLADIUM.getType());
        CODE_TO_METAL_TYPE.put("Au99.99", MetalTypeEnum.GOLD.getType());
        CODE_TO_METAL_TYPE.put("Au(T+D)", MetalTypeEnum.GOLD.getType());
        CODE_TO_METAL_TYPE.put("Ag(T+D)", MetalTypeEnum.SILVER.getType());
        CODE_TO_METAL_TYPE.put("Pt99.95", MetalTypeEnum.PLATINUM.getType());
        CODE_TO_METAL_TYPE.put("XAU", MetalTypeEnum.GOLD.getType());
        CODE_TO_METAL_TYPE.put("XAG", MetalTypeEnum.SILVER.getType());
        CODE_TO_METAL_TYPE.put("XAP", MetalTypeEnum.PLATINUM.getType());
        CODE_TO_METAL_TYPE.put("XPD", MetalTypeEnum.PALLADIUM.getType());
        CODE_TO_METAL_TYPE.put("GC", MetalTypeEnum.GOLD.getType());
        CODE_TO_METAL_TYPE.put("SI", MetalTypeEnum.SILVER.getType());
        CODE_TO_METAL_TYPE.put("PL", MetalTypeEnum.PLATINUM.getType());
        CODE_TO_METAL_TYPE.put("PA", MetalTypeEnum.PALLADIUM.getType());
        CODE_TO_METAL_TYPE.put("au_gj", MetalTypeEnum.GOLD.getType());
        CODE_TO_METAL_TYPE.put("ag_gj", MetalTypeEnum.SILVER.getType());
        CODE_TO_METAL_TYPE.put("pt_gj", MetalTypeEnum.PLATINUM.getType());
        CODE_TO_METAL_TYPE.put("pd_gj", MetalTypeEnum.PALLADIUM.getType());
    }

    @KafkaListener(
            topics = "#{@rtjDataPriceConsumer.getTopicName()}",
            groupId = "#{@rtjDataPriceConsumer.getGroupId()}",
            errorHandler = "rtjDataKafkaListenerErrorHandler",
            containerFactory = "rtjDataKafkaListenerContainerFactory")
    public void onMessage(RtjDataPriceMessage message) {
        handleData(message);
    }

    @Bean("rtjDataKafkaListenerErrorHandler")
    public KafkaListenerErrorHandler rtjDataKafkaListenerErrorHandler() {
        return (message, exception) -> {
            log.error("RtjData Error processing record: " + message, exception);
            return message;
        };
    }

    public String getTopicName() {
        return kafkaTopicProperties.getRtjdata().getName();
    }

    public String getGroupId() {
        return kafkaTopicProperties.getRtjdata().getGroupId();
    }

    private void handleData(RtjDataPriceMessage message) {
        if (message.getData() == null) {
            return;
        }

        // 收集需要缓存的价格数据
        List<LivePriceRespDTO> cachePriceList = new ArrayList<>(CACHE_CODES.size());
        // 收集所有需要发送WebSocket的价格数据
        List<LivePriceRespDTO> wsPriceList = new ArrayList<>(CODE_TO_METAL_TYPE.size());

        message.getData().forEach((key, priceData) -> {
            Integer metalType = CODE_TO_METAL_TYPE.get(priceData.getCode());
            
            // 创建基础价格数据对象
            LivePriceRespDTO priceRespDTO = new LivePriceRespDTO();
            priceRespDTO.setMetalCode(priceData.getCode());

            priceRespDTO.setReclaimPrice(StringUtils.isNotBlank(priceData.getBidprice()) ? new BigDecimal(priceData.getBidprice()) : null);
            priceRespDTO.setSalePrice(StringUtils.isNotBlank(priceData.getAskprice()) ? new BigDecimal(priceData.getAskprice()) : null);
            priceRespDTO.setHighPrice(StringUtils.isNotBlank(priceData.getHigh()) ? new BigDecimal(priceData.getHigh()) : null);
            priceRespDTO.setLowPrice(StringUtils.isNotBlank(priceData.getLow()) ? new BigDecimal(priceData.getLow()) : null);
            priceRespDTO.setOpenPrice(null);
            priceRespDTO.setYesyPrice(null);
            
            // 根据代码决定是否需要缓存
            if (CACHE_CODES.contains(priceData.getCode()) && metalType != null) {
                // 只为缓存数据设置metalName和metalType
                MetalTypeEnum metal = MetalTypeEnum.valueOf(metalType);
                priceRespDTO.setMetalName(metal.getTag());
                priceRespDTO.setMetalType(metal.getType());
                cachePriceList.add(priceRespDTO);
            }
            
            // 所有数据都添加到WebSocket发送列表
            wsPriceList.add(priceRespDTO);
        });

        // 批量写入缓存
        if (!cachePriceList.isEmpty()) {
            try {
                livePriceApi.asyncBatchCachePriceData(cachePriceList);
            } catch (Exception e) {
                log.error("批量缓存价格数据异常", e);
            }
        }

        // 批量发送WebSocket消息
        try {
            webSocketSenderApi.send(LIVE_MESSAGE_PRICE, JSONObject.toJSONString(wsPriceList));
        } catch (Exception e) {
            log.error("批量发送WebSocket价格消息异常", e);
        }
    }
} 